
package main

import (
	"github.com/valyala/fasthttp"
	//"strings"
	"strings"
	"gateway/http"
	"gateway/conf"
	"gateway/lib"
	"sync"
	"os"
	"strconv"
	//"gateway/utils/aliyunlog"
	//"gateway/plugins"
	"time"
     _ "net/http/pprof"
     rawhttp "net/http"
	//"encoding/json"
	//"github.com/json-iterator/go"
	"github.com/json-iterator/go"
	//"encoding/json"
	//rawlog "log"
	//"github.com/rcrowley/go-metrics"
	"github.com/funlake/gopkg/utils/log"
	"github.com/funlake/gopkg/utils"
	"gateway/types"
)
var (
	maxRequestCount int32 = 0
	maxFloodRequest = int32(conf.GetConfig().GetEnvInt("max_flood_reqests"))
	port = ":"+conf.GetConfig().GetEnvString("port")
	lru = lib.NewLru()
	json = jsoniter.ConfigCompatibleWithStandardLibrary
)
func init(){
	if conf.GetConfig().GetEnv() != "prod" {
		go func() {
			rawhttp.ListenAndServe("localhost:6060", nil)
		}()
	}
	println("=========================================")
	println("             _                           ")
	println("  __ _  __ _| |_ _____      ____ _ _   _ ")
	println(" / _` |/ _` | __/ _ \\ \\ /\\ / / _` | | | |")
	println(`| (_| | (_| | ||  __/\ V  V / (_| | |_| |
 \__, |\__,_|\__\___| \_/\_/ \__,_|\__, |
 |___/                             |___/
`)
	println("=========================================")
	//go metrics.Log(metrics.DefaultRegistry, 1 * time.Second, rawlog.New(os.Stderr, "metrics: ", rawlog.Lmicroseconds))
}
func main() {
	log.Success("listening on port %s",port)
	request := http.NewRequest()
	err := fasthttp.ListenAndServe(port, func(ctx *fasthttp.RequestCtx) {
		//ctx.Write([]byte("hello world"))
		curpath := utils.ByteToStr(ctx.Path())
		switch  curpath{
			case "/":
				http.ServeJson(ctx,0,"success")
			default:
				allow := false
				//默认需要配置上这个环境变量
				allowPaths := os.Getenv("GATEWAY_MERGE_ALLOW_PATH")
				//未限制则允许任意路由访问(危险!!!)
				if allowPaths == "" {
					allow = true
				} else{
					//如有限制则判定当前path是否在允许列表内，防止redis被刷爆
					for _,path := range strings.Split(allowPaths,",") {
						if path == curpath {
							allow = true
							break
						}
					}
				}
				if allow {
					//handleDefaultProxy(request,ctx)
					handleMergeProxy(request, ctx)
				}else{
					http.ServeJson(ctx,403,"非法请求(网关阻断)")
				}
		}
	})
	log.Warning("%s",err)
}
func getRouteConfigFromPath(path string) (string,error){
	return lib.NewCache().GetMultiUpstream(strings.Trim(path,"/"))
}

func parseUpstream(upstream string) (vhost string,service string){
	vs := strings.SplitN(strings.TrimSpace(upstream),"/",2)
	path := ""
	if len(vs) > 1 {
		path = vs[1]
	}
	return strings.TrimRight(vs[0], "/"), path
}

type MergeItemData  struct {
	Request JsonRequest `json:"request"`
	Response interface{} `json:"response"`
	Dur time.Duration `json:"duration"`
	CacheKey string `json:"cachekey"`
}
type JsonRequest struct {
	Api string `json:"api"`
	Method  string `json:"method"`
	Respcode string `json:"respcode"`
	Params string  `json:"params"`
	Block bool `json:"block"`
	Cache bool `json:"cache"`
	Default interface{} `json:"default"`
	CacheKey string `json:"cachekey"`
}
type BatchResult struct {
	Code string `json:"code"`
	Data []MergeItemData `json:"data"`
}
//func (b MergeItemData) MarshalJSON()([]byte, error){
//	return []byte(strconv.QuoteToASCII(b.Response.(string))),nil
//}
func handleMergeProxy(request *http.Request,ctx *fasthttp.RequestCtx )  {
	var wg = &sync.WaitGroup{}
	routeCfg,cfgErr :=  getRouteConfigFromPath(utils.ByteToStr(ctx.Path()))
	var pd []JsonRequest
	err := json.Unmarshal(utils.StrToByte(routeCfg),&pd)

	if cfgErr != nil || routeCfg == "" {
		http.ServeJson(ctx,conf.REQUEST_MERGE_PARSE_ERROR,"["+utils.ByteToStr(ctx.Path())+"] 路由未配置")
		return
	}
	if err != nil {
		http.ServeJson(ctx,conf.REQUEST_MERGE_PARSE_ERROR,"请求json参数格式不正确:"+err.Error())
		return
	}
	var Results []MergeItemData
	var ResLock sync.Mutex
	//domain,_ := conf.GetConfig().Get("app").GetString("domain","")
	for _,v := range pd {
		wg.Add(1)
		//atomic.AddInt32(&maxRequestCount,1)
		utils.WrapGo(func(v JsonRequest,wg *sync.WaitGroup) func() {
			return func(){
				//log.Warning("xxx")
				defer wg.Done()
				//defer atomic.AddInt32(&maxRequestCount,-1)
				//vhost, service := parseUpstream(v.Api)
				var(
					gparam = ""
					pparam = ""
					method = "POST"
				)
				if v.Method != ""{
					method = v.Method
				}
				//if len(v.Params) > 0{
				if strings.ToLower(method) == "get" {
					//gparam = v.Api + "?" + v.Params
					gparam = utils.ByteToStr(ctx.Request.RequestURI())
					if v.Params != "" {
						//gparam = utils.StrConcat(gparam,"&",v.Params)
						gparam += "&" + v.Params
					}
				} else{
					pparam = utils.ByteToStr(ctx.Request.Body())
					if v.Params != "" {
						//pparam = utils.StrConcat(pparam,"&",v.Params)
						pparam += "&" + v.Params
					}
				}

				dur := time.Duration(0)
				proxyContext,_ := types.NewProxyContext(ctx)

				cacheKey := proxyContext.GetJwtIss() + ":" + v.CacheKey
				proxyContext.QueryString = gparam
				proxyContext.PostData = utils.StrToByte(pparam)
				proxyContext.Mapkey = v.Api
				proxyContext.Vhost,proxyContext.Service = parseUpstream(v.Api)


				if cacheVal := lru.Get(cacheKey);cacheVal != nil && false{
					//log.Success("%s",cacheVal.Value().(interface{}))
					ResLock.Lock()
					defer ResLock.Unlock()
					Results = append(Results, MergeItemData{
						v,
						cacheVal.Value().(interface{}),
						dur,
						cacheKey,
					})
				} else{
					flag, code, msg, dur := request.ProxyHttpRequest(proxyContext, ctx)
					ResLock.Lock()
					defer ResLock.Unlock()
					if !flag {
						//this.Fail(code,msg)
						Results = append(Results, MergeItemData{
							v,
							&http.CommonResponse{
								strconv.Itoa(code),
								msg,
								nil,
							},
							dur,
							cacheKey,
						})

					}else{
						var sr  map[string] interface{}
						json.Unmarshal(utils.StrToByte(msg),&sr)
						//beego.Warning(sr)
						Results = append(Results, MergeItemData{
							v,
							sr,
							dur,
							cacheKey,
						})
					}
				}
			}
		}(v,wg),"merge_request")
	}
	wg.Wait()

	finalResult,_:= utils.JSONMarshal(BatchResult{
		"0",
		Results,
	})
//	log.Error("%s",Results)
	ctx.Write(filterResponse(finalResult))
}
//业务相关过滤
func filterResponse(response []byte) []byte{
	type R struct{
		Code string `json:"code"`
		Data []map[string] interface{} `json:"data"`
	}
	var(
		m R
		finalCode = "ok"
		finalMsg  = ""
		finalData  = make(map[string] interface{})
		finalDur = ""
	)
	//json encode返回数据
	err := json.Unmarshal(response,&m)
	if err != nil {
		log.Error("%s", err)
	}
	//log.Warning("%s",m)
	//对返回数据进行二次封装
	for _,res := range m.Data{
		r  := make(map[string] interface{})
		var resp map[string] interface{}
		//返回数据
		//fixed : 发现后端接口居然有http:200返回body为空的情况,直接导致网关panic,- -#
		vailReturn := true
		if res["response"] == nil {
			vailReturn = false
		}else {
			resp = res["response"].(map[string]interface{})
			if _,ok := resp["code"];!ok{
				vailReturn = false
			}
		}
		if !vailReturn{
			resp = make(map[string] interface{})
			resp["code"] = float64(conf.REQUEST_EMPTY_RESULT)
			resp["msg"]  = "empty result or not valid json"
			resp["data"] = ""
		}
		//请求原参
		req := res["request"].(map[string] interface{})
		//cachekey

		cachekey := res["cachekey"].(string)
		code := resp["code"].(string)
		respcode := req["respcode"].(string)

		finalDur = strconv.Itoa(int(res["duration"].(float64))/1000000) + "ms"
		//log.Warning(req["api"].(string)+"返回: data:%s,msg:%s",resp["data"],resp["msg"])
		r[respcode] = []string{}
		//log.Warning(cachekey)
		if code == "ok" {
			//成功返回
			r[respcode] = resp["data"]
			//设置缓存是高频动作,可直接1小时缓存
			//正常返回才缓存
			//log.Error("返回cachekey : %d",cachekey)
			if req["cache"].(bool) && cachekey != ""{
				if lruCache := lru.Get(cachekey); lruCache == nil || lruCache.Expired() {
					log.Success("do not setting cache:%s",cachekey)
					//lru.Set(cachekey, r[respcode], time.Hour*1)
				}
			}
			finalMsg = finalMsg + ",success("+respcode+":"+finalDur+")"
			//finalMsg = utils.StrConcat(finalMsg,",success(",respcode,":",finalDur,")")
		}else{
			//异常返回
			//go aliyunlog.Write("gateway","route_proxy_business_error",aliyunlog.Content{utils.ByteToStr(reqlog) ,"fail"},true)
			go log.Error("businiss:gateway,error:route_proxy_business_error,api:%s,reason:%s",req["api"],resp["msg"])
			finalMsg  = finalMsg +"," +resp["msg"].(string)+ "(" + respcode + ":"+finalDur+")"
			//finalMsg  = utils.StrConcat(finalMsg ,resp["msg"].(string), "(" , respcode , ":",finalDur,")")
			r[respcode] = req["default"]
			//触发全局fatal code
			if isGlobalFatalCode(code) && req["block"].(bool){
				finalCode = code
				//go log.Warning("business:gateway,operate:route_fatal_error,content:%s",reqlog)
			//触发网关报错
			}else if isGateWayBlockCode(code){
				//finalMsg =  finalMsg +"," +resp["msg"].(string) + "(" + respcode + ")"
				//r[respcode] = req["default"]
				//go log.Warning("business:gateway,operate:route_gateway_error,content:%s",reqlog)
				//go aliyunlog.Write("gateway","route_general_error",aliyunlog.Content{utils.ByteToStr(reqlog) ,"ok"},false)
			//业务普通报错
			} else {
				//finalMsg =  finalMsg +"," +resp["msg"].(string) + "(" + respcode + ")"
				//go aliyunlog.Write("gateway","route_general_error",aliyunlog.Content{utils.ByteToStr(reqlog) ,"ok"},false)
				//go log.Warning("business:gateway,operate:route_general_error,content:%s",reqlog)
				//一般性错误，赋予默认值
				//r[respcode] = req["default"]
			}
			if req["cache"].(bool){
				if lruCache:=lru.Get(cachekey);lruCache != nil{
					//log.Success("Get value from last successfully cached thing :%s",lruCache)
					//r[respcode] = lruCache.Value().(interface{})
				}
			}
		}
		finalData[respcode] = r[respcode]
		//log.Success("%s",finalData[respcode])
	}
	type Res struct {
		Code string `json:"code"`
		Msg string `json:"msg"`
		Data interface{} `json:"data"`
	}
	//var finalWrap = make([]map[string] interface{},1)
	//finalWrap[0] = finalData
	finalResult,_ := utils.JSONMarshal(Res{
		finalCode,
		 finalMsg,
		finalData,
	})
	return finalResult
}

func isGlobalFatalCode(code string) bool {
	var match  = false
	busFatalCode := os.Getenv("GATEWAY_MERGE_FATAL_CODE")
	if busFatalCode != "" {
		for _, v := range strings.Split(busFatalCode, ",") {
			//vv,_ := strconv.Atoi(v)
			match = (match || (code == v))
		}
	}
	return match
}

func isGateWayBlockCode(code string) bool{
	//match := (code == conf.REQUEST_ROUTE_ERROR ||
	//	code == conf.REQUEST_SERVICE_TEMPORARY_DOWN ||
	//	code == conf.REQUEST_MAX_REACH_ERROR ||
	//	code == conf.REQUEST_TIMEOUT ||
	//	code == conf.REQUEST_RATE_LIMIT ||
	//	code == conf.REQUEST_SERVICE_TEMPORARY_DOWN)
	return false
}